package io.kiki.sba.registry.store.jdbc.elector;

import io.kiki.sba.registry.common.model.elector.Lock;
import io.kiki.sba.registry.store.api.config.DefaultCommonConfig;
import io.kiki.sba.registry.store.api.elector.AbstractLeaderElector;
import io.kiki.sba.registry.store.api.elector.Leader;
import io.kiki.sba.registry.store.api.elector.Role;
import io.kiki.sba.registry.store.api.meta.RecoverConfig;
import io.kiki.sba.registry.store.jdbc.config.MetaElectorConfig;
import io.kiki.sba.registry.store.jdbc.constant.TableEnum;
import io.kiki.sba.registry.store.jdbc.domain.FollowCompeteLockDomain;
import io.kiki.sba.registry.store.jdbc.mapper.LockMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;


public class MetaJdbcLeaderElector extends AbstractLeaderElector implements RecoverConfig {

    public static final String lockName = "META-MASTER";
    private static final Logger logger = LoggerFactory.getLogger(MetaJdbcLeaderElector.class);

    @Autowired
    LockMapper lockMapper;

    @Autowired
    MetaElectorConfig metaElectorConfig;

    @Autowired
    DefaultCommonConfig defaultCommonConfig;

    public static Leader leaderFrom(Lock lock) {
        return calcLeaderInfo(lock.getLockOwner(), lock.getGmtModifiedUnixMillis(), lock.getGmtModifiedUnixMillis(), lock.getDuration());
    }

    /**
     * start elect, return current leader
     */
    @Override
    protected Leader doElect() {
        Lock lock = lockMapper.queryLock(defaultCommonConfig.getClusterId(tableName()), lockName);

        /** compete and return leader */
        if (lock == null) {
            return competeLeader(defaultCommonConfig.getClusterId(tableName()));
        }

        Role role = amILeader(lock.getLockOwner()) ? Role.leader : Role.follower;
        if (role == Role.leader) {
            lock = onLeaderWorking(lock, myself());
        } else {
            lock = onFollowWorking(lock, myself());
        }
        Leader result = leaderFrom(lock);
        logger.info("meta role : {}, leaderInfo: {}", role, result);
        return result;
    }


    private Leader competeLeader(String dataCenter) {
        Lock lock = new Lock(dataCenter, lockName, myself(), metaElectorConfig.getLockExpireDuration());
        try {
            // throw exception if insert fail
            lockMapper.competeLockOnInsert(lock);
            // compete finish.
            lock = lockMapper.queryLock(dataCenter, lockName);

            logger.info("meta: {} compete success, become leader.", myself());
        } catch (Throwable t) {
            // compete leader error, query current leader
            lock = lockMapper.queryLock(dataCenter, lockName);
            logger.info("meta: {} compete error, leader is: {}.", myself(), lock.getLockOwner());
        }
        return leaderFrom(lock);
    }


    @Override
    protected Leader doQuery() {
        Lock lock = lockMapper.queryLock(defaultCommonConfig.getClusterId(tableName()), lockName);
        if (lock == null) {
            return Leader.HAS_NO_LEADER;
        }

        return leaderFrom(lock);
    }

    private Lock onLeaderWorking(Lock lock, String myself) {

        try {
            /** as leader, do heartbeat */
            lockMapper.ownerHeartbeat(lock);
            logger.info("leader heartbeat: {}", myself);
            return lockMapper.queryLock(lock.getDataCenter(), lock.getLockName());
        } catch (Throwable t) {
            logger.error("leader:{} heartbeat error.", myself, t);
        }
        return lock;
    }

    public Lock onFollowWorking(Lock lock, String myself) {
        /** as follow, do compete if lock expire */
        if (lock.expire()) {
            logger.info("lock expire: {}, meta elector start: {}", lock, myself);
            lockMapper.competeLockOnUpdate(new FollowCompeteLockDomain(lock.getDataCenter(), lock.getLockName(), lock.getLockOwner(), lock.getGmtModified(), myself, lock.getDuration(), lock.getTerm(), lock.getTermDuration()));
            Lock newLock = lockMapper.queryLock(lock.getDataCenter(), lock.getLockName());
            logger.info("elector finish, new lock: {}", lock);
            return newLock;
        }
        return lock;
    }

    @Override
    public String tableName() {
        return TableEnum.distribute_lock.getTableName();
    }
}
